package com.ktjy.rabbitmqconsumer7002;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.ktjy.pojo.LogDTO;
import com.ktjy.pojo.MqttMessage;
import com.ktjy.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class EMQXConsumer1 {
    private static Channel channel;

    public static void main(String[] args) {
        Connection connection = null;
        try {
            // 通过 RabbitMQUtil 工具类获取连接
            connection = RabbitMQUtil.newConnection();
            channel = connection.createChannel();
            // 修改队列名为 file_log_queue
            String queueName = "emqx_temp_queue";
            // 声明队列，设置为持久化
            channel.queueDeclare(queueName, true, false, false, null);

            // 监听队列，设置为手动确认消息（autoAck设为false）
            channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
                /**
                 * 重写 handleDelivery 方法，处理消息
                 *
                 * @param consumerTag 消费者标签
                 * @param envelope    消息信封，包含消息的一些元数据
                 * @param properties  消息属性
                 * @param body        消息体
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String json = new String(body);
                    //LogDTO logDTO = JSON.parseObject(json, LogDTO.class);
                    System.out.println("接收到的消息: " + json);
                    ObjectMapper mapper = new ObjectMapper();

                    // 解析外层消息
                    MqttMessage mqttMessage = mapper.readValue(json, MqttMessage.class);

                    // 解析内层负载
                    LogDTO payload = mapper.readValue(mqttMessage.getPayload(), LogDTO.class);

                    System.out.println("用户名: " + mqttMessage.getUsername());
                    System.out.println("主题: " + mqttMessage.getTopic());
                    System.out.println("消息级别: " + payload.getLevel());
                    System.out.println("消息内容: " + payload.getContent());
                    //执行消息的业务逻辑，可以是增加
                    if (!payload.getLevel().equals("1")) {
                        channel.basicAck(envelope.getDeliveryTag(), false);
                        System.out.println("消息已确认");
                    } else {
                        channel.basicNack(envelope.getDeliveryTag(), false, true);
                        System.out.println("消息处理失败，已拒绝并重新入队");
                    }
                    try {
                        Thread.sleep(4000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}